package com.gopay.scheduler.job;

import com.gopay.common.constants.proccode.ProcCodeConstants;
import com.gopay.common.constants.trans.PayBankAsyncOpType;
import com.gopay.common.constants.trans.PgStatus;
import com.gopay.common.constants.trans.Trans00700AssistStatus;
import com.gopay.common.cps.dao.order.CpsGeneralMainOrderQueryDAO;
import com.gopay.common.cps.dao.order.CpsTran00700AssistDAO;
import com.gopay.common.domain.cps.CpsTrans00700Assist;
import com.gopay.common.domain.cps.transfer.Tran00700AsyncVO;
import com.gopay.common.domain.cps.vo.CPSResponse;
import com.gopay.common.exception.GopayException;
import com.gopay.common.order.dao.CommonQueryDAO;
import com.gopay.common.util.DateUtils;
import com.gopay.common.util.NamedThreadFactory;
import com.gopay.remote.order.MainOrderUpdateRemoteService;
import com.gopay.scheduler.GopayBaseJob;
import com.gopay.scheduler.common.CallCPSHelper;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;

/**
 *
 * 00700 异步消费 多线程
 * @author ddh02
 * @Tran00700AsyncJob
 * @2017/7/27 21:27  www.gopay.com.cn Inc.All rights reserved.
 */

public class Tran00700AsyncMTJob extends GopayBaseJob {

    private Logger logger = LoggerFactory.getLogger(Tran00700AsyncMTJob.class);
    private CpsGeneralMainOrderQueryDAO cpsGeneralMainOrderDAO;
    private MainOrderUpdateRemoteService cpsGeneralMainOrderOperClient;

    private static final int QUERYCOUNT = 300;

    @Resource(name = "cpsGeneralMainOrderOperClient")
    public void setCpsGeneralMainOrderOperClient(MainOrderUpdateRemoteService cpsGeneralMainOrderOperClient) {
        this.cpsGeneralMainOrderOperClient = cpsGeneralMainOrderOperClient;
    }

    @Resource(name = "cpsGeneralMainOrderDAO")
    public void setCpsGeneralMainOrderDAO(CpsGeneralMainOrderQueryDAO cpsGeneralMainOrderDAO) {
        this.cpsGeneralMainOrderDAO = cpsGeneralMainOrderDAO;
    }


    @Resource(name = "commonQueryDAO")
    private CommonQueryDAO commonQueryDAO;

    @Resource(name = "cpsTran00700AssistDAO")
    private CpsTran00700AssistDAO tran00700AssistDAO;

    @Resource(name = "callCPSHelper")
    private CallCPSHelper callCPSHelper;

    //线程池数量
    private int threadCount;

    private ThreadPoolExecutor threadPool ;

    private ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor();


    @PostConstruct
    public void PostConstruct() {
        logger.info("初始化线程池数量=" + threadCount);
        threadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(threadCount, new NamedThreadFactory("00700mt"));


        monitor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                String date = DateUtils.getCurrDate();
                int size = threadPool.getQueue().size();
                int threadSize = threadPool.getPoolSize();
                int activeThreadSize = threadPool.getActiveCount();
                long taskCount = threadPool.getTaskCount();
                long completedTaskCount = threadPool.getCompletedTaskCount();
                long largestPoolSize = threadPool.getLargestPoolSize();

                StringBuffer sb = new StringBuffer("线程池参数监控=");
                //sb.append("date=" + date);
                sb.append("size=" + size);
                sb.append(",threadSize=" + threadSize);
                sb.append(",activeThreadSize=" + activeThreadSize);
                sb.append(",taskCount=" + taskCount);
                sb.append(",completedTaskCount=" + completedTaskCount);
                sb.append(",largestPoolSize=" + largestPoolSize);

                logger.info(sb.toString());
            }
        }, 5, 30, TimeUnit.SECONDS);
    }


    public Tran00700AsyncMTJob() {
    }

    @Override
    protected void doJobInternal() {

        long startMillions = System.currentTimeMillis();
        logger.info("00700 异步 多线程 处理开始 ");

        /*String sql = "from CpsTrans00700Assist t where " +
                "t.status= '" + Trans00700AssistStatus.INIT.value +"'" +
                "and t.createdTs >= :date1 order by t.createdTs asc ";

        Map<String, Object> map = new HashMap();

        //date1 多久时间内的数据
        Date date1 = new Date();
        Calendar calendar1 = Calendar.getInstance();
        calendar1.setTime(date1);
        calendar1.add(Calendar.DAY_OF_YEAR, -1);  //减一天
        date1 = calendar1.getTime();
        map.put("date1", date1);

        List<CpsTrans00700Assist> assists = tran00700AssistDAO.findByHql(sql, map, 1, QUERYCOUNT);

        callCps(assists);*/


        //查询 00700 异步辅助表的 状态为00（初始化） 的 一天内的 ，网关流水状态表 为01（成功）的数据
        //订单状态为30000 记账点为null的
        StringBuffer sb = new StringBuffer();
        sb.append("   select a.gopay_order_id, o.trans_dtl_sq  ");
        sb.append("     from CPS_TRANS_00700_ASSIST a  ");
        sb.append("     left join cps_gen_main_order o  ");
        sb.append("       on o.gopay_order_id = a.gopay_order_id  ");
        sb.append("     left join PG_GEN_ORDER_STATUS_DTL os  ");
        sb.append("       on os.id = o.trans_dtl_sq  ");
        sb.append("    where a.status = '" + Trans00700AssistStatus.INIT.value + "'  ");
        sb.append("      and a.created_ts > sysdate - 1  ");
        sb.append("      and os.status = '" + PgStatus.SUCCESS.value + "'  ");
        sb.append("      and o.txn_sta_cd = 30000 ");  //订单状态为30000
        sb.append("      and o.accounting_point is null  ");  //记账点为空
        sb.append("    order by a.created_ts asc  ");

        Map<String, Object> map = new HashMap();
        List<Object[]> list = commonQueryDAO.findBySql(sb.toString(), map, 1, QUERYCOUNT);

        callCps(list);

        long endMiilions = System.currentTimeMillis();
        long result = (endMiilions - startMillions);
        logger.info("00700 异步多线程处理结束 数量{},耗费{}毫秒", list.size(), result);

    }


    class Task implements Runnable {

        // private CpsTrans00700Assist order;

        private String gopayOrderId;

        private String dtlId;

        private CountDownLatch countDownLatch;

        public Task(String orderId, CountDownLatch countDownLatch, String dtlId) {
            //this.order = order;
            this.countDownLatch = countDownLatch;
            this.gopayOrderId = orderId;
            this.dtlId = dtlId;
        }

        @Override
        public void run() {

            try {

                logger.info("开始处理线程池中数据gopayOrderId={},dtlId={}", gopayOrderId, dtlId);
                CpsTrans00700Assist order = tran00700AssistDAO.find(CpsTrans00700Assist.class, gopayOrderId);

                //update by dongdh  注销掉这里的状态
                //order.setLastUpdTs(new Date());
                //更改发送状态
                //order.setStatus(Trans00700AssistStatus.SEND_CPS.value);
                //tran00700AssistDAO.update(order);

                Tran00700AsyncVO vo = new Tran00700AsyncVO();
                vo.setGopayOrderId(gopayOrderId);
                vo.setOpType(PayBankAsyncOpType.CRON_FROM);
                //调用cps

                long start = System.currentTimeMillis();

                CPSResponse cpsResponse = callCPSHelper.callByVOTicket(vo);
                String respCode = cpsResponse.getRespCode();

                long end = System.currentTimeMillis();

                //cps 返回 更改中间表 状态
                if (!StringUtils.equals(ProcCodeConstants.PROC_CODE_100S1000, respCode)) {
                    logger.error("00700调用cps返回码异常respCode={}, gopayOrderId={}, 耗时={}", new Object[]{respCode, gopayOrderId, (end - start)});
                    // order.setStatus(Trans00700AssistStatus.CPS_ERROR.value);
                    // order.setRemark(respCode);
                } else {
                    // order.setStatus(Trans00700AssistStatus.SUCCESS.value);
                    logger.info("线程池中处理cps 调用 成功gopayOrderid=" + gopayOrderId + ", 耗时=" + (end - start));
                }

                //tran00700AssistDAO.update(order);
            } catch (GopayException e) {
                logger.error("调用cps异常gopayOrderId=" + gopayOrderId , e);

            } catch (Exception e) {
                logger.error("调用cps异常gopayOrderId=" + gopayOrderId , e);

            } finally {
                countDownLatch.countDown();
            }
        }
    }


    private void callCps(List<Object[]> assists)  {

        if (!(assists.isEmpty())) {
            logger.info("取到待处理数据条数：" + assists.size());
            CountDownLatch countDownLatch = new CountDownLatch(assists.size());

            logger.info("多线程开始调用");
            for (Object[] order : assists) {
                String orderId = String.valueOf(order[0]);
                String dtlId = String.valueOf(order[1]);
                Task task = new Task(orderId, countDownLatch, dtlId);
                threadPool.submit(task);
            }

            try {
                countDownLatch.await(300, TimeUnit.SECONDS);
                logger.info("多线程调用完毕..");
            } catch (InterruptedException e) {
                logger.error("countdown error", e);
            }
        } else {
            logger.info("00700 取回列表为空...");
        }
    }


    // @Deprecated
    /*private void callCps(List<CpsTrans00700Assist> assists)  {

        if (!(assists.isEmpty())) {
            logger.info("取到待处理数据条数：" + assists.size());
            CountDownLatch countDownLatch = new CountDownLatch(assists.size());

            logger.info("多线程开始调用");
            for (CpsTrans00700Assist order : assists) {
                String orderId = "";
                Task task = new Task(orderId, countDownLatch);
                threadPool.submit(task);
            }

            try {
                countDownLatch.await(60, TimeUnit.SECONDS);
                logger.info("多线程调用完毕..");
            } catch (InterruptedException e) {
                logger.error("countdown error", e);
            }
        } else {
            logger.info("00700 取回列表为空...");
        }
    }*/

    public int getThreadCount() {
        return threadCount;
    }

    public void setThreadCount(int threadCount) {
        this.threadCount = threadCount;
    }
}
